package cn.boat.router.consumer;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.util.*;


/*
* 监控consumerId的变化 和 broker leader的变化 ，以便做负载均衡
*
* */
public class ConsumerWatcherImpl implements ConsumerWatcherInterface {
    private ZooKeeper zooKeeperConnection;
    private static final String CONSUMER_ROOT_PATH = "/boatMQ/consumers";
    private static final String IDS = "/ids";
    // private static final String OFFSETS = "offsets";
    private static final String OWNERS = "/owners";
    private static final String BROKERS = "/boatMQ/brokers";
    private static final String TOPICS = "/topics";
    private static final String PARTITIONS = "/partitions";
    private static final String STATE = "/state";
    private List<OldNewConsumer> myOldNewConsumerList ;
    private HashMap<String, Integer> oldHashMap;
    private HashMap<String, Integer> newHashMap;
    private int oldConsumerCount ;
    private List<OldNewPartitionLeader> oldNewPartitionLeaderList;

    public List<OldNewPartitionLeader> getOldNewPartitionLeaderList() {
        return oldNewPartitionLeaderList;
    }

    private ConsumerMetaData myConsumerMetaData; //当分区leader 发生改变时，监控到次变化，存储的是订阅的topic对应的分区还有 分区leader信息
    public ZooKeeper getZooKeeperConnection() {
        return zooKeeperConnection;
    }

    public List<OldNewConsumer> getOldNewConsumerList() {
        return myOldNewConsumerList;
    }

    public ConsumerMetaData getMyConsumerMetaData() {
        return myConsumerMetaData;
    }

    public void setMyConsumerMetaData(ConsumerMetaData myConsumerMetaData) {
        this.myConsumerMetaData = myConsumerMetaData;
    }

    public ConsumerWatcherImpl(ZooKeeper zooKeeperConnection) {
        myConsumerMetaData = new ConsumerMetaData();
        this.zooKeeperConnection=zooKeeperConnection;
        myOldNewConsumerList = new ArrayList<>();
        oldHashMap = new HashMap<>();
        newHashMap = new HashMap<>();
        oldNewPartitionLeaderList= new ArrayList<>();
    }
/*
* 监控ConsumerId的变化
* */
    public List<OldNewConsumer> balance(String consumerGroupId, String consumerId) throws KeeperException, InterruptedException {
        List<OldNewConsumer> oldNewConsumerList = new ArrayList<>() ;
        List<String> topicList = zooKeeperConnection.getChildren(CONSUMER_ROOT_PATH + "/" + consumerGroupId + OWNERS, false);
        Collections.sort(topicList);
        HashMap<String, List<String>> partitionChangeMap = new HashMap<>();
        HashMap<String, HashMap<String, List<String>>> consumerPartitionChangeMap =new HashMap<>();
        List<String> consumerIdList = new ArrayList<>();
        List<String> partitionList = new ArrayList<>();
        List<String> consumerList =new ArrayList<>();
        int consumerGroupSize = 0;
        consumerList = zooKeeperConnection.getChildren(CONSUMER_ROOT_PATH+"/"+consumerGroupId+IDS,false);
        Collections.sort(consumerList);
        if (consumerList.size()!=0){
            consumerGroupSize = consumerList.size();
            for (String myConsumerId : consumerList) {
                System.out.println(myConsumerId);
                byte[] data = zooKeeperConnection.getData(CONSUMER_ROOT_PATH + "/" + consumerGroupId + IDS + "/" + myConsumerId, false, null);
                System.out.println("数据为"+new String(data));
                String dataStr = new String(data);
                consumerIdList.add(dataStr); // 将consumer id加入到List集合中，以便做负载均衡
            }
        }
        consumerGroupSize = consumerList.size();
        int partitionNumCount =0;
        for (String topic : topicList) {
            List<String> consumerPartionChangeList = new ArrayList<>();  //记录被删除的消费者之前消费过的分区
            partitionList = zooKeeperConnection.getChildren(CONSUMER_ROOT_PATH + "/" + consumerGroupId + OWNERS + "/" + topic, false);
            Collections.sort(partitionList);
            for (String partionNum : partitionList) {
                byte[] partitionWithConsumer = zooKeeperConnection.getData(CONSUMER_ROOT_PATH + "/" + consumerGroupId + OWNERS + "/" + topic + "/" + partionNum, false, null);

                if (Objects.equals(new String(partitionWithConsumer),consumerId)){
                    consumerPartionChangeList.add(partionNum); //被删除的消费者之前消费过的分区
                    partitionNumCount++;// 记录被删除消费者消费过多少分区
                }

            }
            if (consumerPartionChangeList.size() !=0){
                partitionChangeMap.put(topic,consumerPartionChangeList);// 每个topic 对应的分区列表
            }

        }
        consumerPartitionChangeMap.put(consumerId,partitionChangeMap); // 消费者id 对应已经订阅 的topic 以及topic对应的分区列表
        // 把之前消费者所消费的分区平均分给其余消费者
        Set<String> subtopicsSet = partitionChangeMap.keySet();
        for (String topicKey : subtopicsSet) {
            List<String> topicPartitionlist = partitionChangeMap.get(topicKey);
            int topicPartitionlistSize = topicPartitionlist.size();
            if (consumerGroupSize== topicPartitionlistSize){
                System.out.println("进入等于");
                System.out.println(consumerGroupSize);
                for (int i = 0; i < topicPartitionlistSize; i++) {
                    zooKeeperConnection.setData(CONSUMER_ROOT_PATH+"/"+consumerGroupId+OWNERS+"/"+topicKey+"/"+topicPartitionlist.get(i),(consumerIdList.get(i)).getBytes(),-1) ;
                    OldNewConsumer oldNewConsumer = new OldNewConsumer();
                    oldNewConsumer.set(topicKey,topicPartitionlist.get(i),consumerId,consumerIdList.get(i)); //对象 包含 删除的消费者订阅过的topic，分区，老消费者，新消费者
                    oldNewConsumerList.add(oldNewConsumer);
                }
            }
            if (consumerGroupSize>topicPartitionlistSize){
                for (int i = 0; i < topicPartitionlistSize; i++) {
                    zooKeeperConnection.setData(CONSUMER_ROOT_PATH+"/"+consumerGroupId+OWNERS+"/"+topicKey+"/"+topicPartitionlist.get(i),(consumerIdList.get(i)).getBytes(),-1) ;
                    OldNewConsumer oldNewConsumer = new OldNewConsumer();
                    oldNewConsumer.set(topicKey,topicPartitionlist.get(i),consumerId,consumerIdList.get(i)); //对象 包含 删除的消费者订阅过的topic，分区，老消费者，新消费者
                    oldNewConsumerList.add(oldNewConsumer);
                }
            }
            if (consumerGroupSize<topicPartitionlistSize){
                int yuShu = topicPartitionlistSize%consumerGroupSize;
                int averageNum = topicPartitionlistSize / consumerGroupSize;
                for (int i = 0,j=0; i < topicPartitionlistSize; i++,j++) {
                    if (i>0 && ((i+consumerGroupSize)%consumerGroupSize) ==0){
                        averageNum--;
                        j=0;
                    }
                    if (averageNum>0){
                        zooKeeperConnection.setData(CONSUMER_ROOT_PATH+"/"+consumerGroupId+OWNERS+"/"+topicKey+"/"+topicPartitionlist.get(i),(consumerIdList.get(j)).getBytes(),-1) ;
                        OldNewConsumer oldNewConsumer = new OldNewConsumer();
                        oldNewConsumer.set(topicKey,topicPartitionlist.get(i),consumerId,consumerIdList.get(j)); //对象 包含 删除的消费者订阅过的topic，分区，老消费者，新消费者
                        oldNewConsumerList.add(oldNewConsumer);
                    }
                    if (averageNum ==0){
                        if (yuShu>0){
                            zooKeeperConnection.setData(CONSUMER_ROOT_PATH+"/"+consumerGroupId+OWNERS+"/"+topicKey+"/"+topicPartitionlist.get(i),(consumerIdList.get(j)).getBytes(),-1) ;
                            OldNewConsumer oldNewConsumer = new OldNewConsumer();
                            oldNewConsumer.set(topicKey,topicPartitionlist.get(i),consumerId,consumerIdList.get(j)); //对象 包含 删除的消费者订阅过的topic，分区，老消费者，新消费者
                            oldNewConsumerList.add(oldNewConsumer);
                            yuShu--;
                        }
                    }
                }

            }
        }
        // 找到挂掉的消费者所消费的分区，将该分区告知新的消费者，以便到broker可以继续之前的offest消费
        return oldNewConsumerList;
    }
    @Override
    public void watchConsumerId(String consumerGroupId, String consumerId) throws Exception {
        List<String> consumerNameList = zooKeeperConnection.getChildren(CONSUMER_ROOT_PATH + "/" + consumerGroupId+IDS, false);
        for (String consumerName : consumerNameList) {
            byte[] data = zooKeeperConnection.getData(CONSUMER_ROOT_PATH + "/" + consumerGroupId + IDS+ "/" +consumerName, false, null);
            oldHashMap.put(new String(data),10);
        }
        Watcher watcher = new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                try {
                    if (watchedEvent.getType() == Event.EventType.NodeChildrenChanged){
                        // consumer节点发生变化，触发负载均衡
                        List<String> consumerListMy = zooKeeperConnection.getChildren(CONSUMER_ROOT_PATH + "/" + consumerGroupId + IDS, false);
                        if (oldConsumerCount<consumerListMy.size()){
                            List<String> consumerNameList = zooKeeperConnection.getChildren(CONSUMER_ROOT_PATH + "/" + consumerGroupId+IDS, false);
                            for (String consumerName : consumerNameList) {
                                byte[] data = zooKeeperConnection.getData(CONSUMER_ROOT_PATH + "/" + consumerGroupId  +IDS+"/" + consumerName, false, null);
                                oldHashMap.put(new String(data),10);
                            }
                            System.out.println("监测到有consumer加入，已经将consumer元数据重新获取，请查收");
                            GetConsumerMetaDataImpl getConsumerMetaDataMy = new GetConsumerMetaDataImpl();
                            ConsumerMetaData consumerMetaData = getConsumerMetaDataMy.getConsumerMetaData(consumerGroupId, consumerId, zooKeeperConnection);
                            myConsumerMetaData = consumerMetaData;
                            System.out.println("consumer元数据信息已经得到请查收");
                            oldConsumerCount = consumerListMy.size();
                            zooKeeperConnection.getChildren(CONSUMER_ROOT_PATH+"/"+consumerGroupId+IDS,this);
                        }else {

                            // 判断是哪个消费者挂掉
                            List<String> consumerNameList = zooKeeperConnection.getChildren(CONSUMER_ROOT_PATH + "/" + consumerGroupId+IDS, false);
                            for (String consumerName : consumerNameList) {
                                byte[] data = zooKeeperConnection.getData(CONSUMER_ROOT_PATH + "/" + consumerGroupId  +IDS+ "/" +consumerName, false, null);
                                newHashMap.put(new String(data),10);
                            }
                            String overConsumerName ="xxx";
                            Set<String> set = oldHashMap.keySet();
                            for (String consumerN : set) {
                                if (!newHashMap.containsKey(consumerN)){
                                    overConsumerName = consumerN;
                                    break;
                                }
                            }
                            System.out.println("监听到consumergroup:"+consumerGroupId+"下的节点发生删除，现在将消费者之前消费的信息做重新分配");
                            myOldNewConsumerList = balance(consumerGroupId,overConsumerName);
                            oldHashMap =newHashMap;
                            System.out.println("已经获取旧消费者和新消费者对应的topic和partition对应关系，请通过getOldNewConsumerList()方法查收 并让新消费者去broker继续读取旧消费者的offest");
                            // 测试
                            System.out.println("长度"+myOldNewConsumerList.size());
                            for (OldNewConsumer oldNewConsumer : myOldNewConsumerList) {
                                System.out.println("是否执行");
                                System.out.println("OldConsumer:"+oldNewConsumer.getOldConsumer()+" NewConsumer:"+oldNewConsumer.getNewConsumer()+"主题："+oldNewConsumer.getTopics()+"分区:"+oldNewConsumer.getPartitionid());
                            }

                            GetConsumerMetaDataImpl getConsumerMetaDataMy = new GetConsumerMetaDataImpl();
                            ConsumerMetaData consumerMetaData = getConsumerMetaDataMy.getConsumerMetaData(consumerGroupId, consumerId, zooKeeperConnection);
                            myConsumerMetaData = consumerMetaData;
                            System.out.println("consumer元数据信息已经得到请查收");
                            oldConsumerCount = consumerListMy.size();
                            // watcher 机制是一次性的，必须重新注册watcher
                            zooKeeperConnection.getChildren(CONSUMER_ROOT_PATH+"/"+consumerGroupId+IDS,this);
                        }

                    }
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        };
        List<String> consumerListMy = zooKeeperConnection.getChildren(CONSUMER_ROOT_PATH + "/" + consumerGroupId + IDS, false);
        oldConsumerCount = consumerListMy.size();
        System.out.println(oldConsumerCount);
        zooKeeperConnection.getChildren(CONSUMER_ROOT_PATH+"/"+consumerGroupId+IDS,watcher);
    }
    /*
     * 监控TopicsBrokerLeader（订阅了多个topic）的变化
     * */
    @Override
    public void watchIntrestingTopicsBrokerLeaderChange(String consumerGroupId, String consumerId, List<String> topicsList) throws Exception {
        for (String topic : topicsList) {
            watchIntrestingTopicsBrokerLeaderChange(consumerGroupId,consumerId,topic);
        }
    }


    /*
     * 监控BrokerLeader如果被删除，则等待两秒中，等待分区leader重新选举
     * */
    public void watchbrokerLeader(String consumerGroupId, String consumerId, String topic, String partitionNum) throws Exception {
        GetConsumerMetaDataImpl getConsumerMetaDataMy1 = new GetConsumerMetaDataImpl();
        myConsumerMetaData = getConsumerMetaDataMy1.getConsumerMetaData(consumerGroupId, consumerId, zooKeeperConnection);
        Watcher watcher = new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                try {
                    if (watchedEvent.getType() == Event.EventType.NodeDataChanged){
                        System.out.println("监听到"+watchedEvent.getPath()+"有变化，已经更换分区leader");
                        GetConsumerMetaDataImpl getConsumerMetaDataMy = new GetConsumerMetaDataImpl();

                        ConsumerMetaData consumerMetaData = getConsumerMetaDataMy.getConsumerMetaData(consumerGroupId, consumerId, zooKeeperConnection);
                        List<TopicWithPartitionData> topicWithPartitionDataList1 = consumerMetaData.getTopicWithPartitionDataList();
                        List<TopicWithPartitionData> topicWithPartitionDataListMy = myConsumerMetaData.getTopicWithPartitionDataList();
                        Collections.sort(topicWithPartitionDataList1);
                        Collections.sort(topicWithPartitionDataListMy);
                        for (int i=0;i<topicWithPartitionDataList1.size();i++){
                            String topicJudge = topicWithPartitionDataList1.get(i).getTopic();
                            List<PartitionData> partitionDataList = topicWithPartitionDataList1.get(i).getPartitionDataList();
                            List<PartitionData> partitionDataListMy = topicWithPartitionDataListMy.get(i).getPartitionDataList();
                            Collections.sort(partitionDataList);
                            Collections.sort(partitionDataListMy);
                            for (int j=0;j<partitionDataList.size();j++){

                                if (partitionDataList.get(j).getPartitionLeader()!= partitionDataListMy.get(j).getPartitionLeader()){
                                    OldNewPartitionLeader oldNewPartitionLeader =new OldNewPartitionLeader();
                                    // 消费者id，主题，旧分区leader，新分区leader
                                    oldNewPartitionLeader.set(consumerId,topicJudge,partitionDataListMy.get(j).getPartitionLeader(),partitionDataList.get(j).getPartitionLeader());
                                    oldNewPartitionLeaderList.add(oldNewPartitionLeader);
                                }
                            }

                        }
                        myConsumerMetaData = consumerMetaData;
                        System.out.println("分区leader变化，重新获取新的leader对应关系"+oldNewPartitionLeaderList);
                        // watcher 机制是一次性的，必须重新注册watcher
                        zooKeeperConnection.getData(BROKERS + TOPICS + "/" + topic + PARTITIONS + "/" + partitionNum  + STATE, this, null);

                    }
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        };
        zooKeeperConnection.getData(BROKERS + TOPICS + "/" + topic + PARTITIONS + "/" + partitionNum  + STATE, watcher, null);
    }

    /*
     * 监控TopicsBrokerLeader（订阅了一个topic）的变化
     * */
    @Override
    public void watchIntrestingTopicsBrokerLeaderChange(String consumerGroupId, String consumerId, String topic) throws Exception {
        List<String> partionNumList = zooKeeperConnection.getChildren(BROKERS + TOPICS + "/" + topic + PARTITIONS, false);
        Collections.sort(partionNumList);
        for (String partitionNum : partionNumList) {
            watchbrokerLeader(consumerGroupId,consumerId,topic,partitionNum);
        }
    }


}
